package org.jetlinks.pro.device.service.data;

import lombok.extern.slf4j.Slf4j;
import org.hswebframework.web.api.crud.entity.PagerResult;
import org.hswebframework.web.api.crud.entity.QueryParamEntity;
import org.hswebframework.web.exception.BusinessException;
import org.jetlinks.core.cache.Caches;
import org.jetlinks.core.device.DeviceOperator;
import org.jetlinks.core.device.DeviceProductOperator;
import org.jetlinks.core.device.DeviceRegistry;
import org.jetlinks.core.message.DeviceMessage;
import org.jetlinks.core.message.Headers;
import org.jetlinks.core.metadata.DeviceMetadata;
import org.jetlinks.pro.device.entity.DeviceEvent;
import org.jetlinks.pro.device.entity.DeviceOperationLogEntity;
import org.jetlinks.pro.device.entity.DeviceProperty;
import org.jetlinks.pro.timeseries.query.AggregationData;
import org.reactivestreams.Publisher;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import javax.annotation.Nonnull;
import java.util.Map;
import java.util.function.Function;

/**
 * 默认的设备数据服务,用于对设备的消息数据{@link DeviceMessage}进行持久化操作.
 * <br>
 * 此类不实现具体的存储逻辑,而是使用策略模式,根据不同的产品配置的存储策略来选择对应的策略{@link DeviceDataStoragePolicy}进行处理.
 *
 * <br><br>
 * 可通过配置文件来指定默认的存储策略:
 * <pre>{@code
 * jetlinks:
 *  device:
 *      storage:
 *          default-policy: default-row
 * }</pre>
 * <p>
 * 或者在产品中添加配置{@code storePolicy}来动态指定对应产品下设备数据的存储策略:
 *
 * <pre>{@code
 *  product
 *  .setConfig("storePolicy","default-row")
 *  .then(....)
 * }</pre>
 *
 * @author zhouhao
 * @see DeviceDataStoragePolicy
 * @see DeviceDataService#STORE_POLICY_CONFIG_KEY
 * @since 1.5
 */
@Component
@Slf4j
public class DefaultDeviceDataService implements DeviceDataService {

    //注册中心
    private final DeviceRegistry deviceRegistry;

    //所有支持的策略
    private final Map<String, DeviceDataStoragePolicy> policies = Caches.newCache();

    //默认的存储策略,当产品里没有配置存储策略时,使用默认的策略进行存储
    private final Mono<DeviceDataStoragePolicy> defaultPolicyMono;

    //最新数据管理服务
    private final DeviceLatestDataService latestDataService;

    //存储配置
    private final DeviceDataStorageProperties properties;


    public DefaultDeviceDataService(DeviceRegistry registry,
                                    DeviceDataStorageProperties storeProperties,
                                    ObjectProvider<DeviceDataStoragePolicy> policies,
                                    DeviceLatestDataService latestDataService) {
        this.deviceRegistry = registry;
        this.latestDataService = latestDataService;
        this.properties = storeProperties;
        //添加存储策略支持
        for (DeviceDataStoragePolicy policy : policies) {
            this.policies.put(policy.getId(), policy);
        }
        //初始化默认的存储策略获取规则
        defaultPolicyMono = Mono
            .fromSupplier(() -> this.policies.get(this.properties.getDefaultPolicy()))
            //默认存储策略不支持时,抛出错误
            .switchIfEmpty(Mono.error(() -> new BusinessException("error.storage_policy_does_not_exist", 500, this.properties.getDefaultPolicy())));
    }

    @Override
    public Mono<Void> registerMetadata(@Nonnull String productId, @Nonnull DeviceMetadata metadata) {
        //获取存储策略进行物模型注册
        return this
            .getStoreStrategy(productId)
            .flatMap(policy -> policy.registerMetadata(productId, metadata))
            .then(properties.isEnableLastDataInDb()
                      //调用最新数据服务更新物模型
                      //应该有更好的方式,而不是这里直接调用.
                      ? latestDataService.upgradeMetadata(productId, metadata)
                      : Mono.empty());
    }

    @Override
    public Mono<Void> reloadMetadata(@Nonnull String productId, @Nonnull DeviceMetadata metadata) {
        //获取存储策略进行物模型刷新
        return this
            .getStoreStrategy(productId)
            .flatMap(policy -> policy.reloadMetadata(productId, metadata))
            .then(properties.isEnableLastDataInDb()
                      //调用最新数据服务更新物模型
                      //应该有更好的方式,而不是这里直接调用.
                      ? latestDataService.reloadMetadata(productId, metadata)
                      : Mono.empty());
    }


    private Mono<DeviceDataStoragePolicy> getStoreStrategy(DeviceProductOperator product) {
        return product
            //获取产品配置
            .getConfig(DeviceDataService.STORE_POLICY_CONFIG_KEY)
            .map(conf -> Mono
                .justOrEmpty(policies.get(conf))
                //配置了存储且存储策略不存在
                .switchIfEmpty(Mono.error(() -> new BusinessException("error.storage_policy_does_not_exist", 500, conf)))
            )
            //没有配置存储策略则使用默认存储策略
            .defaultIfEmpty(defaultPolicyMono)
            .flatMap(Function.identity());
    }

    //获取产品的存储策略
    private Mono<DeviceDataStoragePolicy> getStoreStrategy(String productId) {
        return deviceRegistry
            .getProduct(productId)
            .flatMap(this::getStoreStrategy);
    }

    //获取设备使用的存储策略
    private Mono<DeviceDataStoragePolicy> getDeviceStrategy(String deviceId) {
        return deviceRegistry
            .getDevice(deviceId)
            //根据产品获取
            .flatMap(DeviceOperator::getProduct)
            .flatMap(this::getStoreStrategy);
    }

    @Nonnull
    @Override
    public Flux<DeviceProperty> queryEachOneProperties(@Nonnull String deviceId,
                                                       @Nonnull QueryParamEntity query,
                                                       @Nonnull String... properties) {
        return this
            .getDeviceStrategy(deviceId)
            .flatMapMany(strategy -> strategy.queryEachOneProperties(deviceId, query, properties));
    }


    @Nonnull
    @Override
    public Flux<DeviceProperty> queryEachProperties(@Nonnull String deviceId,
                                                    @Nonnull QueryParamEntity query,
                                                    @Nonnull String... properties) {
        return this
            .getDeviceStrategy(deviceId)
            .flatMapMany(strategy -> strategy.queryEachProperties(deviceId, query, properties));
    }

    @Nonnull
    @Override
    public Flux<DeviceProperty> queryTopProperty(@Nonnull String deviceId,
                                                 @Nonnull AggregationRequest request,
                                                 int numberOfTop,
                                                 @Nonnull String... properties) {
        return this
            .getDeviceStrategy(deviceId)
            .flatMapMany(strategy -> strategy.queryTopProperty(deviceId, request, numberOfTop, properties));
    }

    @Nonnull
    @Override
    public Flux<DeviceProperty> queryProperty(@Nonnull String deviceId,
                                              @Nonnull QueryParamEntity query,
                                              @Nonnull String... property) {

        return this
            .getDeviceStrategy(deviceId)
            .flatMapMany(strategy -> strategy.queryProperty(deviceId, query, property));
    }

    @Nonnull
    @Override
    public Flux<DeviceProperty> queryPropertyByProductId(@Nonnull String productId,
                                                         @Nonnull QueryParamEntity query,
                                                         @Nonnull String... property) {
        return this
            .getStoreStrategy(productId)
            .flatMapMany(strategy -> strategy.queryPropertyByProductId(productId, query, property));
    }

    @Override
    public Flux<AggregationData> aggregationPropertiesByProduct(@Nonnull String productId,
                                                                @Nonnull AggregationRequest request,
                                                                @Nonnull DevicePropertyAggregation... properties) {
        return this
            .getStoreStrategy(productId)
            .flatMapMany(strategy -> strategy.aggregationPropertiesByProduct(productId, request.copy(), properties));
    }

    @Override
    public Flux<AggregationData> aggregationPropertiesByDevice(@Nonnull String deviceId,
                                                               @Nonnull AggregationRequest request,
                                                               @Nonnull DevicePropertyAggregation... properties) {
        return this
            .getDeviceStrategy(deviceId)
            .flatMapMany(strategy -> strategy.aggregationPropertiesByDevice(deviceId, request.copy(), properties));
    }

    @Nonnull
    @Override
    public Mono<PagerResult<DeviceProperty>> queryPropertyPage(@Nonnull String deviceId,
                                                               @Nonnull String property,
                                                               @Nonnull QueryParamEntity query) {
        return this
            .getDeviceStrategy(deviceId)
            .flatMap(strategy -> strategy.queryPropertyPage(deviceId, property, query))
            .defaultIfEmpty(PagerResult.empty());
    }

    @Nonnull
    @Override
    public Mono<PagerResult<DeviceProperty>> queryPropertyPage(@Nonnull String deviceId,
                                                               @Nonnull QueryParamEntity query,
                                                               @Nonnull String... property) {
        return this
            .getDeviceStrategy(deviceId)
            .flatMap(strategy -> strategy.queryPropertyPage(deviceId, query, property))
            .defaultIfEmpty(PagerResult.empty());
    }

    @Nonnull
    @Override
    public Mono<PagerResult<DeviceProperties>> queryPropertiesPage(@Nonnull String deviceId,
                                                                   @Nonnull QueryParamEntity query) {
        return this
            .getDeviceStrategy(deviceId)
            .flatMap(strategy -> strategy.queryPropertiesPage(deviceId, query))
            .defaultIfEmpty(PagerResult.empty());
    }

    @Nonnull
    @Override
    public Flux<DeviceProperties> queryProperties(@Nonnull String deviceId,
                                                  @Nonnull QueryParamEntity query) {
        return this
            .getDeviceStrategy(deviceId)
            .flatMapMany(strategy -> strategy.queryProperties(deviceId, query));
    }

    @Nonnull
    @Override
    public Mono<PagerResult<DeviceProperties>> queryPropertiesPageByProduct(@Nonnull String productId, @Nonnull QueryParamEntity query) {
        return this
            .getStoreStrategy(productId)
            .flatMap(strategy -> strategy.queryPropertiesPageByProduct(productId, query))
            .defaultIfEmpty(PagerResult.empty());
    }

    @Nonnull
    @Override
    public Mono<PagerResult<DeviceProperty>> queryPropertyPageByProductId(@Nonnull String productId,
                                                                          @Nonnull QueryParamEntity query,
                                                                          @Nonnull String... property) {
        return this
            .getStoreStrategy(productId)
            .flatMap(strategy -> strategy.queryPropertyPageByProductId(productId, query, property))
            .defaultIfEmpty(PagerResult.empty());
    }

    @Nonnull
    @Override
    public Mono<PagerResult<DeviceProperty>> queryPropertyPageByProductId(@Nonnull String productId,
                                                                          @Nonnull String property,
                                                                          @Nonnull QueryParamEntity query) {
        return this
            .getStoreStrategy(productId)
            .flatMap(strategy -> strategy.queryPropertyPageByProductId(productId, property, query))
            .defaultIfEmpty(PagerResult.empty());
    }

    @Override
    public Mono<PagerResult<DeviceOperationLogEntity>> queryDeviceMessageLog(@Nonnull String deviceId, @Nonnull QueryParamEntity query) {
        return this
            .getDeviceStrategy(deviceId)
            .flatMap(strategy -> strategy.queryDeviceMessageLog(deviceId, query))
            .defaultIfEmpty(PagerResult.empty());
    }

    private void tryUpdateLatestProperties(DeviceMessage message) {
        //是否开启最新数据存储,可以全局配置,也可以在消息头中设置ignoreLatest来忽略最新数据存储
        if (!properties.isEnableLastDataInDb() || message.getHeader("ignoreLatest").isPresent()) {
            return;
        }
        latestDataService.save(message);
    }


    private boolean checkDeviceMessageSavable(DeviceMessage msg) {
        //header中不存在产品ID,可能是直接调用方法保存的
        if (!msg.getHeader(Headers.productId).isPresent()) {
            log.warn("设备消息错误,请检查deviceId对应的设备是否已注册:{}", msg);
            return false;
        }
        return true;
    }

    @Nonnull
    @Override
    public Mono<Void> saveDeviceMessage(@Nonnull DeviceMessage message) {
        if (!checkDeviceMessageSavable(message)) {
            return Mono.empty();
        }
        //尝试更新最新数据
        tryUpdateLatestProperties(message);
        return this
            .getDeviceStrategy(message.getDeviceId())
            .flatMap(strategy -> strategy.saveDeviceMessage(message));
    }

    @Nonnull
    @Override
    public Mono<Void> saveDeviceMessage(@Nonnull Publisher<DeviceMessage> message) {
        return Flux
            .from(message)
            //检查消息是否可以保存
            .filter(this::checkDeviceMessageSavable)
            //尝试更新最新数据
            .doOnNext(this::tryUpdateLatestProperties)
            //根据设备分组,相同的设备使用相同的存储策略进行批量存储
            .groupBy(DeviceMessage::getDeviceId, Integer.MAX_VALUE)
            .flatMap(group -> this
                .getDeviceStrategy(group.key())
                .flatMap(policy -> policy.saveDeviceMessage(group)))
            .then();
    }

    @Nonnull
    @Override
    public Flux<DeviceEvent> queryEvent(@Nonnull String deviceId,
                                        @Nonnull String event,
                                        @Nonnull QueryParamEntity query, boolean format) {

        return this
            .getDeviceStrategy(deviceId)
            .flatMapMany(strategy -> strategy.queryEvent(deviceId, event, query, format));
    }

    @Nonnull
    @Override
    public Mono<PagerResult<DeviceEvent>> queryEventPage(@Nonnull String deviceId,
                                                         @Nonnull String event,
                                                         @Nonnull QueryParamEntity query,
                                                         boolean format) {
        return this
            .getDeviceStrategy(deviceId)
            .flatMap(strategy -> strategy.queryEventPage(deviceId, event, query, format))
            .defaultIfEmpty(PagerResult.empty());
    }

    @Nonnull
    @Override
    public Mono<PagerResult<DeviceEvent>> queryEventPageByProductId(@Nonnull String productId,
                                                                    @Nonnull String event,
                                                                    @Nonnull QueryParamEntity query,
                                                                    boolean format) {
        return this
            .getStoreStrategy(productId)
            .flatMap(strategy -> strategy.queryEventPageByProductId(productId, event, query, format))
            .defaultIfEmpty(PagerResult.empty());
    }
}
